package com.settlement.system.controller;

import com.settlement.system.common.result.Result;
import io.swagger.v3.oas.annotations.Operation;
import io.swagger.v3.oas.annotations.security.SecurityRequirement;
import lombok.RequiredArgsConstructor;
import org.apache.kafka.clients.admin.AdminClient;
import org.apache.kafka.clients.admin.ListTopicsResult;
import org.springframework.kafka.config.KafkaListenerEndpointRegistry;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.listener.MessageListenerContainer;
import org.springframework.util.Assert;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.PathVariable;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

@RestController
@RequestMapping("/api/v1/kafka")
@RequiredArgsConstructor
public class KafkaController {

    private final KafkaTemplate<String, String> kafkaTemplate;

    private final KafkaListenerEndpointRegistry registry;

    private final AdminClient adminClient;

    @Operation(summary = "获取Topics", security = {@SecurityRequirement(name = "Authorization")})
    @GetMapping("/topicNames")
    public Result<?> getTopics() {
        ListTopicsResult listTopics = adminClient.listTopics();
        try {
            return Result.success(listTopics.names().get());
        } catch (Exception e) {
            e.printStackTrace();
            return Result.failed(e.getMessage());
        }
    }

    @Operation(summary = "暂停消费", security = {@SecurityRequirement(name = "Authorization")})
    @GetMapping("/stop/{consumerId}")
    public Result<?> stop(@PathVariable String consumerId) {
        MessageListenerContainer listenerContainer = registry.getListenerContainer(consumerId);
        Assert.notNull(listenerContainer, "没有找到对应的消费者");
        // 暂停监听
        listenerContainer.pause();
        return Result.success();
    }

    @Operation(summary = "开始消费", security = {@SecurityRequirement(name = "Authorization")})
    @GetMapping("/start/{consumerId}")
    public Result<?> start(@PathVariable String consumerId) {
        MessageListenerContainer listenerContainer = registry.getListenerContainer(consumerId);
        Assert.notNull(listenerContainer, "没有找到对应的消费者");
        if (!listenerContainer.isRunning()) {
            // 先启动
            listenerContainer.start();
        }
        // 将其恢复
        listenerContainer.resume();
        return Result.success();
    }
}
